package com.chukun.flink.stream.operator.base;

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;

/**
 * @author chukun
 * @version 1.0.0
 * @description keyby 操作符
 * @createTime 2022年05月10日 00:06:00
 */
public class KeyOperator {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Tuple2<Integer,Integer>> streamSource = env.fromCollection(new ArrayList<Tuple2<Integer,Integer>>(){
            {
                add(new Tuple2<>(1,11));
                add(new Tuple2<>(1,22));
                add(new Tuple2<>(10,33));
                add(new Tuple2<>(5,55));
                add(new Tuple2<>(5,55));
            }
        });

        DataStream<Tuple2<Integer, Integer>> mapStream = streamSource.keyBy(new KeySelector<Tuple2<Integer, Integer>, Integer>() {

            @Override
            public Integer getKey(Tuple2<Integer, Integer> tuple2) throws Exception {
                return tuple2.f1;
            }
        });

        mapStream.print("输出结果");

        // 触发程序执行
        env.execute("keyby stream");
    }
}
